import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * https://segmentfault.com/a/1190000037714804
 *
 * Reactor 模式
 * @author songxulin
 * @date 2023/9/19 17:05
 * @version 1.0
 */
public class SimpleNIOThreadHttpServer {

    public static void main(String[] args) throws IOException {
        //创建一个选择器组   传递选择器组的大小 决定使用多少选择器来实现
        SelectorGroup selectorGroup = new SelectorGroup(16);
        //开启一个服务端管道
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        //开启一个服务端专用的选择器
        Selector selector = Selector.open();
        //设置非阻塞
        serverSocketChannel.configureBlocking(false);
        //创建一个连接器
        Acceptor acceptor = new Acceptor(serverSocketChannel, selectorGroup);
        //将服务端通道注册到服务端选择器上  这里会绑定一个新连接接入器
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, acceptor);
        //绑定端口
        serverSocketChannel.bind(new InetSocketAddress(8080));
        //启动处理器
        new Reactor(selector).run();
    }

    private static void handleAccept(SelectionKey key, Selector selector) throws IOException {
        ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
        SocketChannel clientChannel = serverChannel.accept();
        clientChannel.configureBlocking(false);
        clientChannel.register(selector, SelectionKey.OP_READ);
//        System.out.println(count.getAndIncrement());
    }

    private static void handleRead(SelectionKey key) throws IOException {
        SocketChannel channel = (SocketChannel) key.channel();
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        StringBuilder request = new StringBuilder();

        int bytesRead;
        while ((bytesRead = channel.read(buffer)) > 0) {
            buffer.flip();
            while (buffer.hasRemaining()) {
                request.append((char) buffer.get());
            }
            buffer.clear();
        }
        System.out.println("Accepted connection from " +key.toString());
        String httpRequest = request.toString();
        System.out.println("Received request:\n" + httpRequest);

        // 处理HTTP请求，这里可以编写你的业务逻辑
        try {
            Thread.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 响应客户端
        String httpResponse = "HTTP/1.1 200 OK\r\n\r\nHello, World!";
        ByteBuffer responseBuffer = ByteBuffer.wrap(httpResponse.getBytes());
        while (responseBuffer.hasRemaining()) {
            channel.write(responseBuffer);
        }
        channel.close();
    }


    /**
     * Nio事件处理器
     *
     * @author huangfu
     * @date
     */
    static class MyNioEventLoop implements Runnable {
        static final ByteBuffer ALLOCATE = ByteBuffer.allocate(128);
        private final Selector selector;
        private final LinkedBlockingQueue<Runnable> linkedBlockingQueue;
        public MyNioEventLoop(Selector selector) {
            this.selector = selector;
            linkedBlockingQueue = new LinkedBlockingQueue<>();
        }

        public Selector getSelector() {
            return selector;
        }

        public LinkedBlockingQueue<Runnable> getLinkedBlockingQueue() {
            return linkedBlockingQueue;
        }

        //忽略  hashCode和eques

        /**
         * 任务处理器
         */
        @Override
        public void run() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    //进行事件选择  这里我们只处理读事件
                    if (selector.select() > 0) {
                        Set<SelectionKey> selectionKeys = selector.selectedKeys();
                        Iterator<SelectionKey> iterator = selectionKeys.iterator();
                        //处理读事件
                        while (iterator.hasNext()) {
                            SelectionKey next = iterator.next();
                            iterator.remove();
                            if (next.isAcceptable()) {
                                // 处理新的连接请求
                                handleAccept(next, selector);
                            } else if (next.isReadable()) {
                                // 处理已连接的Socket的读取请求
                                handleRead(next);
                            }
//                            if (next.isReadable()) {
//                                SocketChannel channel = (SocketChannel) next.channel();
//                                int read = channel.read(ALLOCATE);
//                                if(read > 0) {
//                                    handleAccept
//                                    System.out.printf("线程%s【%s】发来消-息:",Thread.currentThread().getName(), channel.getRemoteAddress());
//                                    System.out.println(new String(ALLOCATE.array(), StandardCharsets.UTF_8));
//                                }else if(read == -1) {
//                                    System.out.println("连接断开");
//                                    channel.close();
//                                }
//                                ALLOCATE.clear();
//                            }
                        }
                        selectionKeys.clear();
                    }else {
                        //处理异步任务  进行注册
                        while (!linkedBlockingQueue.isEmpty()) {
                            Runnable take = linkedBlockingQueue.take();
                            //异步事件执行
                            take.run();
                        }
                    }
                } catch (IOException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 选择器组
     *
     * @author huangfu
     * @date 2021年3月12日09:44:37
     */
    static class SelectorGroup {
        private final List<MyNioEventLoop> SELECTOR_GROUP = new ArrayList<>(8);
        private static final int AVAILABLE_PROCESSORS = Runtime.getRuntime().availableProcessors();
        private final static AtomicInteger IDX = new AtomicInteger();

        /**
         * 初始化选择器
         * @param count 处理器数量
         * @throws IOException 异常欣喜
         */
        public SelectorGroup(int count) throws IOException {

            for (int i = 0; i < count; i++) {
                Selector open = Selector.open();
                MyNioEventLoop myNioEventLoop = new MyNioEventLoop(open);
                SELECTOR_GROUP.add(myNioEventLoop);
            }
        }

        public SelectorGroup() throws IOException {
            this(AVAILABLE_PROCESSORS << 1);
        }

        /**
         * 轮询获取一个选择器
         * @return 返回一个选择器
         */
        public MyNioEventLoop next(){
            int andIncrement = IDX.getAndIncrement();
            int length = SELECTOR_GROUP.size();

            return SELECTOR_GROUP.get(Math.abs(andIncrement % length));
        }
    }
    /**
     * @author huangfu
     * @date
     */
    static class ThreadContext {
        /**
         * 记录当前使用过的选择器
         */
        public static final Set<MyNioEventLoop> RUN_SELECT = new HashSet<>();
    }

    /**
     * 连接器
     *
     * @author huangfu
     * @date 2021年3月12日10:15:37
     */
    static class Acceptor implements Runnable {
        private final ServerSocketChannel serverSocketChannel;
        private final SelectorGroup selectorGroup;

        public Acceptor(ServerSocketChannel serverSocketChannel, SelectorGroup selectorGroup) {
            this.serverSocketChannel = serverSocketChannel;
            this.selectorGroup = selectorGroup;
        }


        @Override
        public void run() {
            try {
                SocketChannel socketChannel = serverSocketChannel.accept();
                MyNioEventLoop next = selectorGroup.next();

                //向队列追加一个注册任务
                next.getLinkedBlockingQueue().offer(() -> {
                    try {
                        //客户端注册为非阻塞
                        socketChannel.configureBlocking(false);
                        //注册到选择器 关注一个读事件
                        socketChannel.register(next.getSelector(), SelectionKey.OP_READ);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
                //唤醒对应的任务，让其处理异步任务
                next.getSelector().wakeup();


                System.out.println("检测到连接：" + socketChannel.getRemoteAddress());
                //当当前选择器已经被使用过了  就不再使用了，直接注册就行了
                if (ThreadContext.RUN_SELECT.add(next)) {
                    //启动任务
                    new Thread(next).start();
                }


            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 反应器
     *
     * @author huangfu
     * @date 2021年3月12日10:15:14
     */
    static class Reactor implements Runnable {
        private final Selector selector;

        public Reactor(Selector selector) {
            this.selector = selector;
        }

        @Override
        public void run() {
            try {
                System.out.println("服务启动成功");
                while (!Thread.currentThread().isInterrupted()) {
                    //d等待连接事件
                    selector.select();
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        SelectionKey next = iterator.next();
                        iterator.remove();
                        //进行数据分发
                        dispatch(next);
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        /**
         * 将新连接分发到新连接接入器
         * @param next 所有事件主键
         */
        private void dispatch(SelectionKey next) {
            Runnable attachment = (Runnable) next.attachment();
            if(attachment!=null) {
                attachment.run();
            }
        }
    }
}

